package com.ruoyi.base.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqttServer {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqttServer.class);

    /* 订阅者客户端对象 */
    private MqttClient subscribeClient;

    public MqttClient getSubscribeClient() {
        return subscribeClient;
    }

    /**
     * 发布者客户端对象
     * 这里订阅者和发布者的MqttClient对象分别命名是为了让发布者和订阅者分开，
     * 如果订阅者和发布者都用一个MqttClient链接对象，则会出现两方都订阅了某个主题后，
     * 谁发送了消息，都会自己接收到自己发的消息，所以分开写，里面主要就是回调类的设置setCallback
     * */
    private MqttClient publishClient;

    /* 主题对象 */
    public MqttTopic mqttTopic;

    /* 消息内容对象 */
    public MqttMessage mqttMessage;



    @Autowired
    private MqttConfig mqttConfig;


    /**
     * 发送消息
     * @param topic 主题
     * @param data 消息内容
     * @param qos 消息级别
     */
    public void sendMQTTMessage(String topic, String data, int qos) {

        try {
            publishClient = publishConnect();

            mqttTopic = publishClient.getTopic(topic);

            mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setPayload(data.getBytes());


            //发送消息
            try {
                //把消息发送给对应的主题
                MqttDeliveryToken mqttDeliveryToken = mqttTopic.publish(mqttMessage);
                mqttDeliveryToken.waitForCompletion();

                //检查发送是否成功
                boolean flag = mqttDeliveryToken.isComplete();

                String tip = "";
                if (flag) {
                    tip = "给主题 "+mqttTopic.getName() + " 发布消息成功。" +
                            "消息内容是："+new String(mqttMessage.getPayload());
                } else {
                    tip = "给主题 "+mqttTopic.getName() + " 发布消息失败。";
                }
                LOGGER.info(tip);

            } catch (MqttException e) {
                LOGGER.info(e.toString());
            }

        } catch (Exception e) {
            LOGGER.info(e.toString());
            e.printStackTrace();
        }

    }


    //以某个消息级别订阅某个主题
    public void init(String topic, int qos) {
        subscribeConnect();
        try {
            subscribeClient.subscribe(topic, qos);
        } catch (MqttException e) {
            LOGGER.info(e.getMessage(), e);
        }
    }

    //取消订阅主题
    public void unionInit(String topic) {
        subscribeConnect();
        try {
            subscribeClient.unsubscribe(topic);
        } catch (MqttException e) {
            LOGGER.info(e.getMessage(), e);
        }
    }




    //发布者客户端和服务端建立连接
    public MqttClient publishConnect() {

        try {
            //防止重复创建MQTTClient实例
            if (publishClient==null) {
                //先让客户端和服务器建立连接，MemoryPersistence设置clientid的保存形式，默认为以内存保存
                publishClient = new MqttClient(mqttConfig.getHost(), mqttConfig.getClientid(), new MemoryPersistence());
                //发布消息不需要回调连接
            }

            MqttConnectOptions options = getOptions();

            //连接状态
            if (!publishClient.isConnected()) {
                publishClient.connect(options);
            }else {
                //连接失败，在重新连接一次
                publishClient.disconnect();
                publishClient.connect(getOptions(options));
            }

        } catch (MqttException e) {
            LOGGER.info(e.toString());
        }
        return publishClient;
    }

    /**
     * 订阅端的链接方法，关键是回调类的设置，要对订阅的主题消息进行处理
     * 断线重连方法，如果是持久订阅，重连时不需要再次订阅
     * 如果是非持久订阅，重连是需要重新订阅主题 取决于options.setCleanSession(true);
     * true为非持久订阅
     */
    public void subscribeConnect() {
        try {

            if (subscribeClient ==null) {

                subscribeClient = new MqttClient(mqttConfig.getHost(), mqttConfig.getClientid2(), new MemoryPersistence());
                //订阅者需要回调类，发布不需要
                subscribeClient.setCallback(new SubscribeCallBack(MqttServer.this));
            }

            MqttConnectOptions options = getOptions();

            if (!subscribeClient.isConnected()) {
                subscribeClient.connect(options);
            }else {
                subscribeClient.disconnect();
                subscribeClient.connect(getOptions(options));
            }
            LOGGER.info("***************连接成功");
        } catch (MqttException e) {
            LOGGER.info(e.getMessage(), e);
        }
    }



//--------------------------------------------------------------------------

    //设置链接
    public MqttConnectOptions getOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(mqttConfig.isCleansession());
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setConnectionTimeout(mqttConfig.getConnectionTimeout());
        //设置心跳
        options.setKeepAliveInterval(mqttConfig.getKeepalive());
        return options;
    }

    public MqttConnectOptions getOptions(MqttConnectOptions options) {

        options.setCleanSession(options.isCleanSession());
        options.setUserName(options.getUserName());
        options.setPassword(options.getPassword());
        options.setConnectionTimeout(options.getConnectionTimeout());
        options.setKeepAliveInterval(options.getKeepAliveInterval());
        return options;
    }

}
